from application.etl.base_etl import BaseEtl
from application.etl.read.mongo_read.nsfc_to_kafka import NsfcToKafka
from application.etl.write.print_write import PrintWriter
from application.models.kafka_pathway_schema import KafkaPathwaySchema
from application.pipelines.nsfc_field_cleaning import NsfcFieldCleaning
from application.pipelines.type_label_pipeline import TypeLablePipeline
import pathway as pw
from application.settings import KAFKA_CONFIG


class NsfcToKafkaEtl(BaseEtl):
    """
    信息消费者类：用于从 MongoDB 读取爬虫资讯数据，经过管道处理后写入 Kafka。

    功能概述
    ----------
    1. 从 MongoDB（通过 `NsfcToKafka`）读取原始资讯数据
    2. 通过 `TypeLablePipeline`、`NsfcFieldCleaning` 等管道进行数据清洗与补充标签
    3. 将清洗后的结构化数据写入 Kafka（JSON 格式）

    属性
    ----------
    pipelines : list
        在 ETL 过程中应用的处理管道列表。
        - TypeLablePipeline("information_nsfc"): 负责补充资讯类型标签
        - NsfcFieldCleaning(): 负责清洗和规范化资讯字段
    """

    pipelines = [TypeLablePipeline("information_nsfc"), NsfcFieldCleaning()]
    topic_name = 'temp5'

    def read(self):
        """
        读取 MongoDB 中的资讯数据。

        使用 Pathway 的 `pw.io.python.read` 方法，
        通过 `NsfcToKafka` 读取器加载数据，并应用 `KafkaPathwaySchema` 进行结构化约束。

        :return: Pathway 表对象，包含 MongoDB 原始资讯数据
        """
        input_data = pw.io.python.read(
            NsfcToKafka(),
            schema=KafkaPathwaySchema,
            mode="static",  # 静态模式
        )
        return input_data

    def write(self, transform_data):
        """
        将处理后的数据写入 Kafka。

        当前实现：
        ----------
        - 使用 Pathway 的 `pw.io.kafka.write`
        - 采用 JSON 格式序列化
        - Kafka 配置从 `KAFKA_CONFIG['pathway_producer']` 中读取
        - 目标 topic 为 `temp5`

        可扩展：
        ----------
        - 改为写入 ES 或本地文件
        - 支持多 topic 分发

        :param transform_data: 已完成 transform 的 Pathway 表对象
        """
        # 如果仅需调试，可写入控制台：
        # pw.io.python.write(transform_data, PrintWriter())

        pw.io.kafka.write(
            transform_data,
            rdkafka_settings={
                **KAFKA_CONFIG['pathway_producer'],
                "linger.ms": '1000',  # 每 0.5 秒发送一次批量消息
                "enable.idempotence": 'true',
            },
            topic_name=self.topic_name,
            format="json"
        )
